package com.atguigu.func;

import com.atguigu.pojo.Event;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author WEIYUNHUI
 * @date 2023/6/13 10:39
 *
 * 自定义Source ， 需要实现SourceFunction
 */
public class ClickSource implements SourceFunction<Event> {

    private static Boolean isRunning = true;
    /**
     * 每秒生成一条Event数据
     */
    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        String [] users = {"Zhangs" , "Lisi" , "Tom" , "Jerry" , "Peiqi"};
        String [] urls = {"/home" , "/cart" , "detail" , "pay"} ;
        Random random = new Random( );
        while(isRunning){
            Event event =
                    new Event(users[random.nextInt(users.length)], urls[random.nextInt(urls.length)], System.currentTimeMillis());
            ctx.collect(event);
            //休眠1秒钟
            TimeUnit.SECONDS.sleep(1);
        }
    }

    @Override
    public void cancel() {
        isRunning = false ;
    }
}
